package org.jaychan.live.id.generate.provider.service.impl;

import com.alibaba.fastjson2.JSON;
import jakarta.annotation.Resource;
import org.jaychan.live.id.generate.provider.dao.mapper.IdGenerateMapper;
import org.jaychan.live.id.generate.provider.dao.po.IdGeneratePO;
import org.jaychan.live.id.generate.provider.service.IdGenerateService;
import org.jaychan.live.id.generate.provider.service.bo.LocalSeqIdBO;
import org.jaychan.live.id.generate.provider.service.bo.LocalUnSeqIdBO;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.stereotype.Service;

import java.util.ArrayList;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicLong;


@Service
public class IdGenerateServiceImpl implements IdGenerateService, InitializingBean {

    private static final Logger LOGGER = LoggerFactory.getLogger(IdGenerateServiceImpl.class);
    private static Map<Integer, LocalSeqIdBO> localSeqIdMap = new ConcurrentHashMap<>();

    private static Map<Integer, LocalUnSeqIdBO> localUnSeqIdMap = new ConcurrentHashMap<>();

    private static final float UPDATE_RATE = 0.75f;
    private static final int SEQ_ID = 1;

    private static ThreadPoolExecutor threadPoolExecutor = new ThreadPoolExecutor(8, 16, 3, TimeUnit.SECONDS, new ArrayBlockingQueue<>(1000), new ThreadFactory() {
        @Override
        public Thread newThread(Runnable r) {
            Thread thread = new Thread(r);
            thread.setName("id-generate-thread" + ThreadLocalRandom.current().nextInt(1000));
            return thread;
        }
    });

    private static Map<Integer, Semaphore> semaphoreMap = new ConcurrentHashMap<>();


    @Resource
    private IdGenerateMapper idGenerateMapper;


    @Override
    public Long getUnSeqId(Integer id) {

        if (id == null) {
            LOGGER.error("[getSeqId] id is error,id is {}", id);
            return null;
        }

        LocalUnSeqIdBO localUnSeqIdBO = localUnSeqIdMap.get(id);
        if (localUnSeqIdBO == null) {
            LOGGER.error("[getUnSeqId] id is error,id is {}", JSON.toJSONString(localUnSeqIdBO));
            return null;
        }
        //弹出元素
        Long returnId = localUnSeqIdBO.getIdQueue().poll();
        if (returnId == null) {
            LOGGER.error("[getUnSeqId] returnId is null,id is {}", id);
            return null;
        }

        //元素快用完自动刷新
        this.refreshLocalUnSeqId(localUnSeqIdBO);
        return returnId;
    }


    @Override
    public Long getSeqId(Integer id) {

        if (id == null) {
            LOGGER.error("[getSeqId] id is error,id is {}", id);
            return null;
        }

        LocalSeqIdBO localSeqIdBO = localSeqIdMap.get(id);
        if (localSeqIdBO == null) {
            LOGGER.error("[localSeqIdBO] id is error,id is {}", JSON.toJSONString(localSeqIdBO));
            return null;
        }
        //10000,100150
        //快要跑到150的时候，告诉mysql，去更新这个字段

        //A: 10000 - 100050
        //B: 100050 - 100100
        //A下次去用 100100 - 100150，极端场景下可能有很多IO请求，更新的时候失败，然后重试
        //思路：快用完的时候提前去抢占更新，有点像hashmap的扩容原理

        this.refreshLocalSeqId(localSeqIdBO);
        long returnId = localSeqIdBO.getCurrentNum().getAndIncrement();

        //如果本地id段已经用完，就不能返回新的id了，是为了给调用方用的，怕超了，有个保护机制
        //10000 - 100050 ，已经用完到10050了
        if (returnId > localSeqIdBO.getNextThreshold()) {
            LOGGER.error("[getSeqId] id is over limit,id is {}", JSON.toJSONString(localSeqIdBO));

            //同步去刷新还是returnNull呢
            //返回null，避免大量请求同时进来，导致数据库压力过大
            //因为dubbo底层实现也是线程池，避免同时很多的请求并行的情况，大胆一点，返回个null
            //初期设计就要跟业务系统说好
            return null;
        }
        return returnId;
    }

    /**
     * 刷新本地ID段
     *
     * @param localSeqIdBO
     */
    private void refreshLocalSeqId(LocalSeqIdBO localSeqIdBO) {
        //相差区间
        long step = localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart();
        if (localSeqIdBO.getNextThreshold() - localSeqIdBO.getCurrentStart() > step * UPDATE_RATE) {
            Semaphore semaphore = semaphoreMap.get(localSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("[semaphore] id is error,id is {}", localSeqIdBO.getId());
                return;
            }

            //直接acquire可能会堵塞,tryAcquire提高性能
            boolean acquired = semaphore.tryAcquire();
            if (acquired) {
                LOGGER.info("开始尝试进行本地id段同步");
                //2.可能有很多请求同时过来，要保证线程池里面只有一个任务在执行，怎么做？用信号量,相当于限流
                //1.如果很多线程同步执行，很多网络IO，性能慢
                threadPoolExecutor.execute(() -> {
                    try {
                        IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localSeqIdBO.getId());
                        tryUpdateMySqlRecord(idGeneratePO);
                    } catch (Exception e) {
                        LOGGER.error("[refreshLocalSeqId] error is ", e);
                    } finally {
                        semaphoreMap.get(localSeqIdBO.getId()).release();
                    }
                    LOGGER.info("本地有序id段同步完成，id is {}", localSeqIdBO.getId());
                });
            }


        }

    }


    private void refreshLocalUnSeqId(LocalUnSeqIdBO localUnSeqIdBO) {
        Long begin = localUnSeqIdBO.getCurrentStart();
        Long end = localUnSeqIdBO.getNextThreshold();
        int remainSize = localUnSeqIdBO.getIdQueue().size();
        if ((end - begin) * 0.25 > remainSize) {

            Semaphore semaphore = semaphoreMap.get(localUnSeqIdBO.getId());
            if (semaphore == null) {
                LOGGER.error("[semaphore] id is error,id is {}", localUnSeqIdBO.getId());
                return;
            }
            //直接acquire可能会堵塞,tryAcquire提高性能
            boolean acquired = semaphore.tryAcquire();
            if (acquired) {
                threadPoolExecutor.execute(() -> {
                    try {
                        IdGeneratePO idGeneratePO = idGenerateMapper.selectById(localUnSeqIdBO.getId());
                        tryUpdateMySqlRecord(idGeneratePO);
                        LOGGER.info("无序id段同步完成，id is {}", localUnSeqIdBO.getId());
                    } catch (Exception e) {
                        LOGGER.error("[refreshLocalUnSeqId] error is ", e);

                    } finally {
                        semaphoreMap.get(localUnSeqIdBO.getId()).release();
                    }
                });
            }
        }
    }


    //初始化map，有点像postConstruct
    @Override
    public void afterPropertiesSet() throws Exception {
        List<IdGeneratePO> idGeneratePOS = idGenerateMapper.selectAll();
        for (IdGeneratePO idGenerate : idGeneratePOS) {
            tryUpdateMySqlRecord(idGenerate);
            semaphoreMap.put(idGenerate.getId(), new Semaphore(1));
        }

    }


    private void tryUpdateMySqlRecord(IdGeneratePO idGenerate) {
        //考虑到多线程问题，更新失败要重试
        int updateRow = idGenerateMapper.updateNewIdCountAndVersion(idGenerate.getId(), idGenerate.getVersion());

        if (updateRow > 0) {
            localIdHandler(idGenerate);
            return;
        }


        //重试机制
        for (int i = 0; i < 3; i++) {
            idGenerate = idGenerateMapper.selectById(idGenerate.getId());
            updateRow = idGenerateMapper.updateNewIdCountAndVersion(idGenerate.getId(), idGenerate.getVersion());

            if (updateRow > 0) {
                localIdHandler(idGenerate);
                //更新成功跳出循环
                return;
            }
        }
        throw new RuntimeException("表id段占用失败，重试3次都无法成功，竞争过于激烈。id is " + idGenerate.getId());


    }

    private void localIdHandler(IdGeneratePO idGenerate) {
        //边界开始值
        long currentStart = idGenerate.getCurrentStart();
        //边界结束值
        long nextThreshold = idGenerate.getNextThreshold();
        //内存中当前值
        long currentNum = currentStart;

        //有序
        if (idGenerate.getIsSeq() == SEQ_ID) {

            LocalSeqIdBO localSeqIdBO = new LocalSeqIdBO();
            AtomicLong atomicLong = new AtomicLong(currentNum);
            localSeqIdBO.setId(idGenerate.getId());
            localSeqIdBO.setCurrentNum(atomicLong);
            localSeqIdBO.setCurrentStart(currentStart);
            localSeqIdBO.setNextThreshold(nextThreshold);

            localSeqIdMap.put(localSeqIdBO.getId(), localSeqIdBO);
        } else {

            LocalUnSeqIdBO localUnSeqIdBO = new LocalUnSeqIdBO();
            localUnSeqIdBO.setId(idGenerate.getId());
            localUnSeqIdBO.setCurrentStart(currentStart);
            localUnSeqIdBO.setNextThreshold(nextThreshold);

            Long begin = localUnSeqIdBO.getCurrentStart();
            Long end = localUnSeqIdBO.getNextThreshold();
            ArrayList<Long> idList = new ArrayList<>();
            for (Long i = begin; i < end; i++) {
                idList.add(i);
            }

            //打乱顺序
            Collections.shuffle(idList);
            ConcurrentLinkedQueue<Long> idQueue = new ConcurrentLinkedQueue<>();
            idQueue.addAll(idList);
            localUnSeqIdBO.setIdQueue(idQueue);

            localUnSeqIdMap.put(localUnSeqIdBO.getId(), localUnSeqIdBO);


        }

    }
}
